package demo.db.elasticSearch.highClient;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import demo.db.mysql.vo.SlowSqlStatistic;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.BucketOrder;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedLongTerms;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedStringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms.Bucket;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.ParsedSum;
import org.elasticsearch.search.aggregations.metrics.ParsedValueCount;
import org.elasticsearch.search.aggregations.metrics.SumAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.ValueCountAggregationBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import com.alibaba.fastjson.JSON;

import demo.db.redis.vo.RedisSlowLogStatistic;
import demo.db.redis.vo.Slowlog;

/**
 * 聚合分析
 */
public class AggregationDemo {

    private static Logger logger = LogManager.getRootLogger();

    private RestHighLevelClient client;

    @Before
    public void setUp() {
        client = RestHighLevelClientDemo.getClient();
    }

    @After
    public void after() {
        if (client != null) {
            try {
                client.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    @Test
    public void test2() {
        long now = System.currentTimeMillis();
        long start = now - TimeUnit.DAYS.toMillis(100);
        try {
            List<SlowSqlStatistic> list = slowSqlStatistic(start, now);
            list.forEach(System.err::println);
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    /**
     * 统计应用请求DB的慢SQL耗时及数量
     *
     * @param start
     * @param end
     * @return
     * @throws IOException
     */
    public List<SlowSqlStatistic> slowSqlStatistic(long start, long end) throws IOException {
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        // 执行时间
        boolQueryBuilder
                .filter(QueryBuilders.rangeQuery("timestamp").from(start).to(end));
        // 执行结果
        boolQueryBuilder.must(QueryBuilders.termQuery("resResult", "success"));


        SumAggregationBuilder sumByResponseTime = AggregationBuilders.sum("sumByResponseTime").field("responseTime");
        TermsAggregationBuilder appTermsBuilder = AggregationBuilders.terms("groupByApp").field("app");
        appTermsBuilder.subAggregation(sumByResponseTime);
        TermsAggregationBuilder dbUrlTermsBuilder = AggregationBuilders.terms("groupByDbUrl").field("dbUrl");
        dbUrlTermsBuilder.subAggregation(appTermsBuilder);

        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(boolQueryBuilder);
        searchSourceBuilder.aggregation(dbUrlTermsBuilder);

        System.err.println("source: " + searchSourceBuilder);
        //  慢SQL日志索引
        SearchRequest searchRequest = new SearchRequest("slow_sql_index_alias");
        searchRequest.source(searchSourceBuilder);
        // 查询
        SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
        System.out.println(searchResponse);
        // 统计
        Aggregations aggregations = searchResponse.getAggregations();
        ParsedStringTerms groupByDbUrlTerms = (ParsedStringTerms) aggregations.get("groupByDbUrl");

        List<SlowSqlStatistic> list = new ArrayList<>();
        // 结果解析
        for (Terms.Bucket dbUrlBucket : groupByDbUrlTerms.getBuckets()) {
            String dbUrl = dbUrlBucket.getKeyAsString();
            Aggregations appAggregation = dbUrlBucket.getAggregations();
            ParsedStringTerms appTerms = appAggregation.get("groupByApp");
            for (Terms.Bucket bucket : appTerms.getBuckets()) {
                String app = bucket.getKeyAsString();
                Aggregations appAggregations = bucket.getAggregations();
                ParsedSum parsedSum = (ParsedSum) appAggregations.get("sumByResponseTime");
                double cost = parsedSum.getValue();
                long count = bucket.getDocCount();
                SlowSqlStatistic vo = new SlowSqlStatistic();
                vo.setCnt(count);
                vo.setApp(app);
                vo.setCost(cost);
                vo.setDbUrl(dbUrl);
                list.add(vo);
            }
        }
        return list;
    }

    /**
     * 分组查询各个实例、各个应用的慢命令数量和总耗时
     *
     * @return 各个实例的慢命令数量及其耗时
     * @throws IOException
     */
    @Test
    public void slowSqlStatistic() throws IOException {
        long[] timestampRange = null;

        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        // 执行结果
        boolQueryBuilder.must(QueryBuilders.termQuery("resResult", "success"));
        // 执行时间
        if (timestampRange != null && timestampRange.length == 2) {
            boolQueryBuilder
                    .filter(QueryBuilders.rangeQuery("timeStamp").gte(timestampRange[0]).lte(timestampRange[1]));
        }


        SumAggregationBuilder sumByResponseTime = AggregationBuilders.sum("sumByResponseTime").field("responseTime");
        TermsAggregationBuilder appTermsBuilder = AggregationBuilders.terms("groupByApp").field("app");
        appTermsBuilder.subAggregation(sumByResponseTime);
        TermsAggregationBuilder dbUrlTermsBuilder = AggregationBuilders.terms("groupByDbUrl").field("dbUrl");
        dbUrlTermsBuilder.subAggregation(appTermsBuilder);

        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(boolQueryBuilder);
        searchSourceBuilder.aggregation(dbUrlTermsBuilder);
        System.out.println(searchSourceBuilder.toString());

        //  慢SQL日志索引
        String indexName = "slow_sql_index_alias";
        SearchRequest searchRequest = new SearchRequest(indexName);
        searchRequest.source(searchSourceBuilder);
        // 查询
        SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
        System.out.println(searchResponse);
        // 统计
        Aggregations aggregations = searchResponse.getAggregations();
        ParsedStringTerms groupByDbUrlTerms = (ParsedStringTerms) aggregations.get("groupByDbUrl");

        List<SlowSqlStatistic> list = new ArrayList<>();
        for (Bucket dbUrlBucket : groupByDbUrlTerms.getBuckets()) {
            String dbUrl = dbUrlBucket.getKeyAsString();
            Aggregations appAggregation = dbUrlBucket.getAggregations();
            ParsedStringTerms appTerms = appAggregation.get("groupByApp");
            for (Bucket bucket : appTerms.getBuckets()) {
                String app = bucket.getKeyAsString();
                Aggregations appAggregations = bucket.getAggregations();
                ParsedSum parsedSum = (ParsedSum) appAggregations.get("sumByResponseTime");
                double cost = parsedSum.getValue();
                long count = bucket.getDocCount();
                SlowSqlStatistic vo = new SlowSqlStatistic();
                vo.setCnt(count);
                vo.setApp(app);
                vo.setCost(cost);
                vo.setDbUrl(dbUrl);
                list.add(vo);
            }
        }
        list.forEach(System.err::println);
    }


    /**
     * 分组查询各个实例的慢命令数量和总耗时
     *
     * @return 各个实例的慢命令数量及其耗时
     * @throws IOException
     */
    @Test
    public void slowLogStatistic() throws IOException {
        long[] timestampRange = null;
        // 执行时间
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        if (timestampRange != null && timestampRange.length == 2) {
            boolQueryBuilder
                    .filter(QueryBuilders.rangeQuery("timeStamp").gte(timestampRange[0]).lte(timestampRange[1]));
        }
        SumAggregationBuilder sumExecutionTime = AggregationBuilders.sum("sumByExecutionTime").field("executionTime");

        TermsAggregationBuilder portTermsBuilder = AggregationBuilders.terms("groupByPort").field("redisPort");
        portTermsBuilder.subAggregation(sumExecutionTime);

        TermsAggregationBuilder ipTermsBuilder = AggregationBuilders.terms("groupByIp").field("redisIp");
        ipTermsBuilder.subAggregation(portTermsBuilder);

        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(boolQueryBuilder);
        searchSourceBuilder.aggregation(ipTermsBuilder);

        // Redis 慢日志索引
        String indexName = "slow_redis_index.2021-04-10";
        SearchRequest searchRequest = new SearchRequest(indexName);
        searchRequest.source(searchSourceBuilder);
        // 查询
        SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
        // 统计

        Aggregations aggregations = searchResponse.getAggregations();
        ParsedStringTerms ipTerms = (ParsedStringTerms) aggregations.get("groupByIp");

        List<RedisSlowLogStatistic> list = new ArrayList<>();
        for (Terms.Bucket ipBucket : ipTerms.getBuckets()) {
            String ip = ipBucket.getKeyAsString();
            Aggregations ipAggregation = ipBucket.getAggregations();
            ParsedLongTerms portTerms = (ParsedLongTerms) ipAggregation.get("groupByPort");
            for (Terms.Bucket bucket : portTerms.getBuckets()) {
                Number port = bucket.getKeyAsNumber();
                long docCount = bucket.getDocCount();
                Aggregations portAggregations = bucket.getAggregations();
                ParsedSum parsedSum = (ParsedSum) portAggregations.get("sumByExecutionTime");
                double sumByExecutionTime = parsedSum.getValue();
                RedisSlowLogStatistic vo = new RedisSlowLogStatistic();
                vo.setCnt((int) docCount);
                vo.setIp(ip);
                vo.setPort(port.intValue());
                vo.setExecutionTime(sumByExecutionTime);
                list.add(vo);
            }
        }
        list.forEach(System.err::println);
    }

    @Test
    public void sumTimeGroupByIpPort() throws IOException {
        SumAggregationBuilder sumExecutionTime = AggregationBuilders.sum("sum_executionTime").field("executionTime");

        TermsAggregationBuilder portTermsBuilder = AggregationBuilders.terms("group_by_redisPort").field("redisPort");
        portTermsBuilder.subAggregation(sumExecutionTime);

        TermsAggregationBuilder ipTermsBuilder = AggregationBuilders.terms("group_by_redisIp").field("redisIp");
        ipTermsBuilder.subAggregation(portTermsBuilder);

        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.aggregation(ipTermsBuilder);

        String indexName = "slow_redis_index.2021-04-10";
        SearchRequest searchRequest = new SearchRequest(indexName);
        searchRequest.source(searchSourceBuilder);
        System.err.println(searchRequest);
        System.err.println("SearchRequest : " + JSON.toJSONString(searchRequest, true));

        SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
        System.err.println(searchResponse);

        Aggregations aggreations = searchResponse.getAggregations();
        ParsedStringTerms aggregationByIp = (ParsedStringTerms) aggreations.get("group_by_redisIp");
        System.out.println("aggregationByIp : " + aggregationByIp);
        for (Bucket bucket : aggregationByIp.getBuckets()) {
            System.err.println("Buket : " + bucket);
            String redisIp = bucket.getKeyAsString();
            Aggregations aggreations2 = bucket.getAggregations();
            ParsedLongTerms aggregationByPort = (ParsedLongTerms) aggreations2.get("group_by_redisPort");
            for (Bucket bucket2 : aggregationByPort.getBuckets()) {
                Number redisPort = bucket2.getKeyAsNumber();
                Aggregations aggreations3 = bucket2.getAggregations();
                ParsedSum parsedSum = (ParsedSum) aggreations3.get("sum_executionTime");
                double value = parsedSum.getValue();
                System.err.println("IP:" + redisIp + " port:" + redisPort + " value : " + value);
            }
        }
        Map<String, Aggregation> groupByIp = aggreations.getAsMap();
        System.err.println("groupByIp : " + groupByIp);
        for (Aggregation aggregation : aggreations) {
            System.err.println("aggregation : " + aggregation);
        }
        for (Entry<String, Aggregation> entry : groupByIp.entrySet()) {
        }
        SearchHit[] searchHits = searchResponse.getHits().getHits();
        List<Slowlog> results = Arrays.stream(searchHits)
                .map(hit -> JSON.parseObject(hit.getSourceAsString(), Slowlog.class)).collect(Collectors.toList());
        results.forEach(System.err::println);
//        EsUtils.printSearchResponse(searchResponse);
    }

    @Test
    public void countByIpPort() throws IOException {
        ValueCountAggregationBuilder valueCountAggregationBuilder = AggregationBuilders.count("documentCnt")
                .field("timeStamp");

        TermsAggregationBuilder portTermsBuilder = AggregationBuilders.terms("group_by_redisPort").field("redisPort");
        portTermsBuilder.subAggregation(valueCountAggregationBuilder);

        TermsAggregationBuilder ipTermsBuilder = AggregationBuilders.terms("group_by_redisIp").field("redisIp");
        ipTermsBuilder.subAggregation(portTermsBuilder);

        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.aggregation(ipTermsBuilder);

        String indexName = "slow_redis_index.2021-04-10";
        SearchRequest searchRequest = new SearchRequest(indexName);
        searchRequest.source(searchSourceBuilder);
        System.err.println(searchRequest);
        System.err.println("SearchRequest : " + JSON.toJSONString(searchRequest, true));

        SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
        System.err.println(searchResponse);

        Aggregations aggreations = searchResponse.getAggregations();
        ParsedStringTerms aggregationByIp = (ParsedStringTerms) aggreations.get("group_by_redisIp");
        System.out.println("aggregationByIp : " + aggregationByIp);
        for (Bucket bucket : aggregationByIp.getBuckets()) {
            System.err.println("Buket : " + bucket);
            String redisIp = bucket.getKeyAsString();
            Aggregations aggreations2 = bucket.getAggregations();
            ParsedLongTerms aggregationByPort = (ParsedLongTerms) aggreations2.get("group_by_redisPort");
            for (Bucket bucket2 : aggregationByPort.getBuckets()) {
                Number redisPort = bucket2.getKeyAsNumber();
                long docCount = bucket2.getDocCount();
                Aggregations aggreations3 = bucket2.getAggregations();
                ParsedValueCount parsedValueCount = (ParsedValueCount) aggreations3.get("documentCnt");
                long value = parsedValueCount.getValue();
                System.err.println(
                        "IP:" + redisIp + " port:" + redisPort + " documentCnt : " + value + " docCount:" + docCount);
            }
        }
        Map<String, Aggregation> groupByIp = aggreations.getAsMap();
        System.err.println("groupByIp : " + groupByIp);
        for (Aggregation aggregation : aggreations) {
            System.err.println("aggregation : " + aggregation);
        }
        for (Entry<String, Aggregation> entry : groupByIp.entrySet()) {
        }
        SearchHit[] searchHits = searchResponse.getHits().getHits();
        List<Slowlog> results = Arrays.stream(searchHits)
                .map(hit -> JSON.parseObject(hit.getSourceAsString(), Slowlog.class)).collect(Collectors.toList());
        results.forEach(System.err::println);
//        EsUtils.printSearchResponse(searchResponse);
    }

    public void demo() {
        try {

            // 1、创建search请求
            // SearchRequest searchRequest = new SearchRequest();
            SearchRequest searchRequest = new SearchRequest("bank");

            // 2、用SearchSourceBuilder来构造查询请求体 ,请仔细查看它的方法，构造各种查询的方法都在这。
            SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();

            sourceBuilder.size(0);

            // 加入聚合
            // 字段值项分组聚合
            TermsAggregationBuilder aggregation = AggregationBuilders.terms("by_age").field("age")
                    .order(BucketOrder.aggregation("average_balance", true));
            // 计算每组的平均balance指标
            aggregation.subAggregation(AggregationBuilders.avg("average_balance").field("balance"));
            sourceBuilder.aggregation(aggregation);

            searchRequest.source(sourceBuilder);

            // 3、发送请求
            SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);

            // 4、处理响应
            // 搜索结果状态信息
            if (RestStatus.OK.equals(searchResponse.status())) {
                // 获取聚合结果
                Aggregations aggregations = searchResponse.getAggregations();
                Terms byAgeAggregation = aggregations.get("by_age");
                logger.info("aggregation by_age 结果");
                logger.info("docCountError: " + byAgeAggregation.getDocCountError());
                logger.info("sumOfOtherDocCounts: " + byAgeAggregation.getSumOfOtherDocCounts());
                logger.info("------------------------------------");
                for (Bucket buck : byAgeAggregation.getBuckets()) {
                    logger.info("key: " + buck.getKeyAsNumber());
                    logger.info("docCount: " + buck.getDocCount());
                    logger.info("docCountError: " + buck.getDocCountError());
                    // 取子聚合
//                    Avg averageBalance = buck.getAggregations().get("average_balance");

//                    logger.info("average_balance: " + averageBalance);
                    logger.info("------------------------------------");
                }
                // 直接用key 来去分组
                /*
                 * Bucket elasticBucket = byCompanyAggregation.getBucketByKey("24"); Avg
                 * averageAge = elasticBucket.getAggregations().get("average_age"); double avg =
                 * averageAge.getValue();
                 */

            }

        } catch (IOException e) {
            logger.error(e);
        }
    }

    public void testAggregation() {
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        TermsAggregationBuilder aggregation = AggregationBuilders.terms("by_company").field("company.keyword");
        aggregation.subAggregation(AggregationBuilders.avg("average_age").field("age"));
        searchSourceBuilder.aggregation(aggregation);
    }
}
